package css.sword.annotation.kafka;

import java.io.InputStream;
import java.util.Properties;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.apache.kafka.clients.producer.KafkaProducer;

/**
 * @project 深圳OA
 * @package zhbg.oaxt.redis.impl
 * @file RedisCacheImpl.java 创建时间:2018年7月23日下午6:38:11
 * @title 标题（要求能简洁地表达出类的功能和职责）
 * @description 描述（简要描述类的职责、实现方式、使用注意事项等）
 * @copyright Copyright (c) 2018 中国软件与技术服务股份有限公司
 * @company 中国软件与技术服务股份有限公司
 * @module 模块: 模块名称
 * @author 史雪涛
 * @reviewer 审核人
 * @version 1.0
 * @history 修订历史（历次修订内容、修订人、修订时间等）
 *
 */
public class OAKafkaImpl extends OAKafkaAbs {
    /**
     * instance
     */
    private static OAKafkaImpl instance;

    /**
     * @constructor 构造方法
     */
    protected OAKafkaImpl() throws Exception {
        super();
    }

    /**
     * @name 中文名称
     * @description 相关说明
     * @time 创建时间:2018年7月23日下午6:40:05
     * @return 1
     * @throws Exception
     * @author 史雪涛
     * @history 修订历史（历次修订内容、修订人、修订时间等）
     */
    public static OAKafkaAbs getInstance() throws Exception {
        if (instance == null) {
            synchronized (OAKafkaImpl.class) {
                if (instance == null) {
                    instance = new OAKafkaImpl();
                }
            }
        }

        return instance;
    }

    /**
     * 
     * @name 中文名称
     * @description 相关说明
     * @time 创建时间:2018年7月23日下午6:40:17
     * @throws Exception
     * @author 史雪涛
     * @history 修订历史（历次修订内容、修订人、修订时间等）
     */
    @SuppressWarnings("rawtypes")
    @Override
    protected void init() throws Exception {
        final InputStream in = OAKafkaImpl.class.getClassLoader().getResourceAsStream("producer.properties");
        final Properties properties = new Properties();
        properties.load(in);
        // 设置kafka
        final GenericObjectPoolConfig config = new GenericObjectPoolConfig();
        config.setMaxTotal(100);// 最大连接数
        config.setMaxIdle(50);// 最大空闲连接数
        config.setMinIdle(25);// 最小空闲连接数
        config.setTestOnBorrow(true);// 在获取连接的时候检查有效性, 默认false
        config.setTestWhileIdle(true);// 在空闲时检查有效性, 默认false
        final OAKafkaPool<KafkaProducer>  pool = new KafkaPool(config,properties);
        this.commands = new OAKafkaCommandsImpl(pool);
    }
}
